package com.mango.ch04;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 分别处理每个文件的mapper
 * 
 * @author Mango
 *
 *         2018年8月2日-上午10:56:56
 */
public class MR_Left_Out_Join_Mapper {

	public static class MR_Left_Out_Join_UserMapper extends Mapper<LongWritable, Text, MyPair, MyPair> {

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, MyPair, MyPair>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			// data: userid locationid
			String[] tokens = value.toString().split(",");
			String locationTD = tokens[1];
			String userID = tokens[0];
			// make sure products arrive at a reducer after location
			MyPair outKey = new MyPair(userID, "1");
			MyPair outValue = new MyPair("L", locationTD);
			// 输出 [(userID, "1"),("L", locationTD)]
			context.write(outKey, outValue);
		}

	}

	public static class MR_Left_Out_Join_TranscationMapper extends Mapper<LongWritable, Text, MyPair, MyPair> {

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, MyPair, MyPair>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			// data: trancationid productid userid quantity amount
			String[] tokens = value.toString().split(",");
			String productTD = tokens[1];
			String userID = tokens[2];
			MyPair outKey = new MyPair(userID, "2");
			MyPair outValue = new MyPair("P", productTD);
			// 输出[(userID, "2"),("P", productTD)]
			context.write(outKey, outValue);
		}

	}
}
